package com.bdi.mqtt_broker.mqtt_protocol;

import com.bdi.mqtt_broker.config.BrokerConfig;
import com.bdi.mqtt_broker.constant.Constants;
import com.bdi.mqtt_broker.model.SessionStore;
import com.bdi.mqtt_broker.service.ChannelManagerService;
import com.bdi.mqtt_broker.service.ISessionStoreService;
import com.bdi.mqtt_broker.service.ISubscribeStoreService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

/**
 * @author WangNing yogehaoren@gmial.com <br>
 * @since 1.0
 * Connect 消息处理器
 */
@Slf4j
public class Connect {

    private final BrokerConfig brokerConfig;

    /**
     * 订阅服务接口
     */
    private final ISubscribeStoreService subscribeStoreService;

    private final ChannelGroup channelGroup;

    private final ChannelManagerService channelManagerService;

    private final ISessionStoreService sessionStoreService;

    public Connect(BrokerConfig brokerConfig, ISubscribeStoreService subscribeStoreService,
                   ISessionStoreService sessionStoreService,
                   ChannelGroup channelGroup, ChannelManagerService channelManagerService){

        this.brokerConfig = brokerConfig;
        this.subscribeStoreService = subscribeStoreService;
        this.channelGroup = channelGroup;
        this.sessionStoreService = sessionStoreService;
        this.channelManagerService = channelManagerService;
    }

    public void processConnect(Channel channel, MqttConnectMessage msg){

        // 消息解码器出现异常
        if (msg.decoderResult().isFailure()) {
            Throwable cause = msg.decoderResult().cause();
            if (cause instanceof MqttUnacceptableProtocolVersionException) {
                // 不支持的协议版本
                MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION, false), null);
                channel.writeAndFlush(connAckMessage);
                channel.close();
                return;
            } else if (cause instanceof MqttIdentifierRejectedException) {
                // 不合格的clientId
                MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false), null);
                channel.writeAndFlush(connAckMessage);
                channel.close();
                return;
            }
            channel.close();
            return;
        }
        // clientId为空或null的情况, 这里要求客户端必须提供clientId, 不管cleanSession是否为1, 此处没有参考标准协议实现
        if (!StringUtils.hasText(msg.payload().clientIdentifier())) {
            MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                    new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                    new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false), null);
            channel.writeAndFlush(connAckMessage);
            channel.close();
            return;
        }

        // 如果会话中已存储这个新连接的clientId, 就关闭之前该clientId的连接
        if (sessionStoreService.containsKey(msg.payload().clientIdentifier())) {
            SessionStore sessionStore = sessionStoreService.get(msg.payload().clientIdentifier());
            // 清除之前绘画的消息
            sessionStoreService.remove(msg.payload().clientIdentifier());
            subscribeStoreService.removeForClient(msg.payload().clientIdentifier());

            try {
                ChannelId channelId = channelManagerService.get(sessionStore.getBrokerId() + "_" + sessionStore.getChannelId());
                if (channelId != null) {
                    Channel previous = channelGroup.find(channelId);
                    if (previous != null) previous.close();
                }
            } catch (Exception e) {
                //e.printStackTrace();
            }
        } else {
            //如果不存在session，则清除之前的其他缓存
            subscribeStoreService.removeForClient(msg.payload().clientIdentifier());
            sessionStoreService.remove(msg.payload().clientIdentifier());
        }
        // 处理遗嘱信息
        SessionStore sessionStore = new SessionStore(brokerConfig.getBrokerId(), msg.payload().clientIdentifier(), channel.id().asLongText());
        // 至此存储会话信息及返回接受客户端连接
        sessionStoreService.put(msg.payload().clientIdentifier(), sessionStore);
        // 存储ClientID
        channel.attr(AttributeKey.valueOf(Constants.CHANNEL_ATTRIBUTE_CLIENT_ID)).set(msg.payload().clientIdentifier());
        boolean sessionPresent = sessionStoreService.containsKey(msg.payload().clientIdentifier());
        MqttConnAckMessage okResp = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, sessionPresent), null);
        channel.writeAndFlush(okResp);
        log.debug("CONNECT - clientId: {}, BrokerId: {}", msg.payload().clientIdentifier(), brokerConfig.getBrokerId());
    }

}
